package com.myflink.day05;

import com.myflink.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

import javax.annotation.Nullable;

/**
 * @author Shelly An
 * @create 2020/9/24 8:48
 * 连续5s水位上升
 */
public class ProcessFunction_Timer {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 4444)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] datas = value.split(",");
                        return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
                    }
                })
                .assignTimestampsAndWatermarks(
                        new AssignerWithPunctuatedWatermarks<WaterSensor>() {
                            private Long maxTs = Long.MIN_VALUE;

                            @Nullable
                            @Override
                            public Watermark checkAndGetNextWatermark(WaterSensor lastElement, long extractedTimestamp) {
                                maxTs = Math.max(maxTs, extractedTimestamp);
                                return new Watermark(maxTs);
                            }

                            @Override
                            public long extractTimestamp(WaterSensor element, long previousElementTimestamp) {
                                return element.getTs() * 1000L;
                            }
                        }
                );

        SingleOutputStreamOperator<String> resultDS = sensorDS
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    private Long triggerTs = 0L;
                    //保存上一次的水位
                    private Integer lastVC = 0;

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        //定时器触发，说明已经满足  连续5s 水位上升
                        out.collect(ctx.getCurrentKey() + "在" + timestamp + "监测到水位连续5s上升");
                        System.out.println(timestamp + "\t定时器触发");
                        //将保存的注册时间清空，后面如果继续上升也是要报警的
                        triggerTs = 0L;
                    }

                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                        //判断上升还是下降
                        if (value.getVc()>lastVC) {
                            //上升
                            //第一条数据来的时候注册定时器（即5s内只注册这一个）
                            if (triggerTs==0) {
                                ctx.timerService().registerEventTimeTimer(value.getTs()*1000+5000);
                                triggerTs = value.getTs()*1000+5000;
                            }
                        }else {
                            //未上升
                            //删除注册的定时器
                            ctx.timerService().deleteEventTimeTimer(triggerTs);
                            //重新注册定时器
                            //ctx.timerService().registerEventTimeTimer(value.getTs() * 1000+5000);
                            // （或把保存的时间清空）
                            triggerTs=0L;
                        }
                        //不管上升还是下降，都要保存水位值，供下条数据使用，进行比较
                        lastVC = value.getVc();
                    }
                });

        resultDS.print("result");

        env.execute();
    }
}
